package com.burning.demo.flink.TableAPIAndSQL;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

// **********************
// BLINK STREAMING QUERY
// **********************
/**
 * Author: Mr.Deng
 * Date: 2018/10/15
 * Desc: 使用flink对指定窗口内的数据进行实时统计，最终把结果打印出来
 * 先在node21机器上执行nc -l -p 9999
 */
public class D1Sql {
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
		EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
		StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
		// or TableEnvironment bsTableEnv = TableEnvironment.create(bsSettings);
		// 注册一个 "Orders" 源表，和 "RubberOrders" 结果表
		bsTableEnv.executeSql("CREATE TABLE user_behavior ("
		        + "  user_id BIGINT,"
		        + "  item_id BIGINT,"
		        + "  category_id BIGINT,"
		        + "  behavior STRING,"
		        + "  ts TIMESTAMP(3)"
		        + ") WITH ("
		        + " 'connector' = 'filesystem',"
		        + " 'path' = 'file:///D:\\data\\a_b_c_d.txt',"
		        + " 'format' = 'csv',"
		        + " 'csv.ignore-parse-errors' = 'true',"
		        + " 'csv.allow-comments' = 'true'"
		        + ")"
		        + "");
	}
}
